use std::{
    collections::hash_map::{Entry, HashMap},
    sync::Arc,
};

use futures::{channel::mpsc, select, FutureExt, SinkExt};

use async_std::{
    io::BufReader,
    net::{TcpListener, TcpStream, ToSocketAddrs},
    prelude::*,
    task,
};

type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
type Sender<T> = mpsc::UnboundedSender<T>;
type Receiver<T> = mpsc::UnboundedReceiver<T>;

#[derive(Debug)]
enum Void {}

pub(crate) fn main() -> Result<()> {
    println!("openning tcpsever: 127.0.0.1:8080...");
    task::block_on(accept_loop("127.0.0.1:8080"))
}

async fn accept_loop(addr: impl ToSocketAddrs) -> Result<()> {
    //mpsc = Multiple Producer Single Consumer
    //创建管道，此管道用来判断连接端口发送的消息属性
    let (broker_sender, broker_receiver) = mpsc::unbounded();

    let broker = task::spawn(broker_loop(broker_receiver));
    //绑定ip进行侦听
    let listener = TcpListener::bind(addr).await?;
    let mut incoming = listener.incoming();
    println!("Successfully opened!");
    //有端口接入的时候触发
    while let Some(stream) = incoming.next().await {
        let stream = stream?;
        println!("Accepting from: {}", stream.peer_addr()?);
        spawn_and_log_error(connection_loop(broker_sender.clone(), stream));
    }
    drop(broker_sender);
    broker.await;
    Ok(())
}

async fn connection_loop(mut broker: Sender<Event>, stream: TcpStream) -> Result<()> {
    let stream = Arc::new(stream);//创建原子指针
    let reader = BufReader::new(&*stream);//创建stream的BufReader
    let mut lines = reader.lines();//按行读取接收缓存

    //创建管道用来判断异步函数是否结束，结束管道释放被销毁触发None
    let (_shutdown_sender, shutdown_receiver) = mpsc::unbounded::<Void>();
    //获取接入网络请求用户名同时异常断开检测
    let name = match lines.next().await {
        None => return Err("peer disconnected immediately".into()),
        Some(line) => line?,
    };
    println!("name:{}",name);

    //创建check用的状态管道
    let (checkname_sender, mut check_receiver) = mpsc::unbounded::<bool>();
    //向broker管道发送新用户数据
    broker
        .send(Event::NewPeer {
            name: name.clone(),
            stream: Arc::clone(&stream),
            shutdown: shutdown_receiver,
            checkname_sender: checkname_sender.clone(),
        })
        .await
        .unwrap();

    //check管道触发获取重命名判断结果
    while let Some(check) = check_receiver.next().fuse().await{
        if check == true{
            println!("name:{}  registered successfully!",name);
            break
        } else {
            println!("[Name repetition]:{} is Occupied!!!",name);
            return Ok(())
        }
    }
    //按行循环接收stream中的数据 
    loop{
        //判断是否断连
        let line = match lines.next().await{
            Some(line)=> line?,
            None=>{
                // print!("user:{} disconnected",name);
                break;
            },
        };
        println!("line:{}",line);

        //按照格式划分数据，格式   目的用户们:消息
        let (dest, msg) = match line.find(':') {
            None => continue,
            Some(idx) => (&line[..idx], line[idx + 1..].trim()),
        };
        println!("dest:{}, msg:{}",dest,msg);

        //再次划分格式，找出所有目的用户  用户，用户
        let dest: Vec<String> = dest
            .split(',')
            .map(|name| name.trim().to_string())
            .collect();
        let msg: String = msg.trim().to_string();

        //向broker管道发送Message数据
        broker
            .send(Event::Message {
                from: name.clone(),
                to: dest,
                msg,
            })
            .await
            .unwrap();
    }

    // while let Some(line) = lines.next().await {
    //     let line = line?;
    //     println!("line:{}",line);
    //     let (dest, msg) = match line.find(':') {
    //         None => continue,
    //         Some(idx) => (&line[..idx], line[idx + 1..].trim()),
    //     };
    //     println!("dest:{}, msg:{}",dest,msg);

    //     let dest: Vec<String> = dest
    //         .split(',')
    //         .map(|name| name.trim().to_string())
    //         .collect();
    //     let msg: String = msg.trim().to_string();

    //     broker
    //         .send(Event::Message {
    //             from: name.clone(),
    //             to: dest,
    //             msg,
    //         })
    //         .await
    //         .unwrap();
    // }
    // println!(".");
    Ok(())
}

//发送触发函数
async fn connection_writer_loop(
    messages: &mut Receiver<String>,
    stream: Arc<TcpStream>,
    mut shutdown: Receiver<Void>,
) -> Result<()> {
    //通过解引用，让stream获得完全所有权
    let mut stream = &*stream;
    //发送个ok让用户端确定连接成功
    stream.write_all(b"ok").await?;
    
    loop {
        //select!等待message或shutdown管道谁先触发执行哪个
        select! {
            //如果数据先接收管道先接收，发送数据
            msg = messages.next().fuse() => match msg {
                Some(msg) => stream.write_all(msg.as_bytes()).await?,
                None =>{
                    println!("None msg");
                    break
                },
            },
            //如果shutdown管道销毁先触发销毁信息，显示断连端口
            void = shutdown.next().fuse() => match void {
                Some(void) => match void {},
                None => {
                    //获取已经断连的ip及其端口
                    println!("{} shutdown!",stream.peer_addr()?);
                    break
                },
            }
        }
    }
    Ok(())
}

//创建枚举，newpeer新用户信息，message信息
#[derive(Debug)]
enum Event {
    NewPeer {
        name: String,
        stream: Arc<TcpStream>,
        shutdown: Receiver<Void>,
        checkname_sender: Sender<bool>,
    },
    Message {
        from: String,
        to: Vec<String>,
        msg: String,
    },
}

//broker管道管理函数
async fn broker_loop(mut events: Receiver<Event>) {
    //此管道，用于函数执行完端口断连从列表删除用
    let (disconnect_sender, mut disconnect_receiver) = mpsc::unbounded::<(String, Receiver<String>)>();
    //创建hashmap，带key的动态容器
    let mut peers: HashMap<String, Sender<String>> = HashMap::new();

    loop {
        //判断是事件信息还是断连信息，select!先触发才执行
        let event = select! {
            //接收到事件触发
            event = events.next().fuse() => match event {
                None => break,
                Some(event) => event,
            },
            //当断连管道触发，从peers中删除断连用户
            disconnect = disconnect_receiver.next().fuse() => {
                let (name, _pending_messages) = disconnect.unwrap();
                assert!(peers.remove(&name).is_some());
                println!("user:{} disconnected!",name);
                continue;
            },
        };

        //通过event的结构判断是需要发送的数据还是添加新网络到HashMap
        match event {
            Event::Message { from, to, msg } => {
                for addr in to {
                    //get_mut通过key查找容器中名字是否存在
                    if let Some(peer) = peers.get_mut(&addr) {
                        let msg = format!("from {}:{}\n", from, msg);
                        //发送msg到目标网络
                        peer.send(msg).await.unwrap();
                    } else {
                        //当目标用户不存在时，向源用户发送消息
                        println!("no name:{} is connecting...",addr);
                        if let Some(peer) = peers.get_mut(&from){
                            let msg = format!("no user named:{}!",from);
                            peer.send(msg).await.unwrap();
                        }
                    }
                }
            }
            Event::NewPeer {
                name,
                stream,
                shutdown,
                mut checkname_sender,
            } => match peers.entry(name.clone()) {//用entry匹配name的key是否存在
                Entry::Occupied(..) => {
                    println!("the name:{} is Occupied!!!",name);
                    //解引用获得可变权
                    let mut stream = &*stream;
                    //发送名字重复信息给用户
                    stream.write_all(b"name_repetition").await.unwrap();

                    //通过管道返回个用户名已占用信息到connetct_loop函数中
                    checkname_sender.send(false).await.unwrap();
                    ()
                },
                Entry::Vacant(entry) => {
                    let (client_sender, mut client_receiver) = mpsc::unbounded();
                    entry.insert(client_sender);//此处给peer的添加，entry(name.clone())括号中的内容name.clone()会作为key
                                                //，client_sender作为内容
                    println!("{:?}",peers);
                    let mut disconnect_sender = disconnect_sender.clone();
                    spawn_and_log_error(async move {
                        checkname_sender.send(true).await.unwrap();
                        let res = connection_writer_loop(&mut client_receiver, stream, shutdown).await;

                        //函数结束时用户断连，发送断连的用户名到断连管道
                        disconnect_sender
                            .send((name, client_receiver))
                            .await
                            .unwrap();
                        res
                    });
                }
            },
        }
        // println!("{:?}",peers);
    }
    //本程序运行不到以下代码，手动退出
    drop(peers);
    drop(disconnect_sender);
    println!("drop!");
    while let Some((_name, _pending_messages)) = disconnect_receiver.next().await {
        println!("name:{}, pending_messages:{:?}", _name, _pending_messages);
    }
}

//打印错误信息
fn spawn_and_log_error<F>(fut: F) -> task::JoinHandle<()>
where
    F: Future<Output = Result<()>> + Send + 'static,
{
    task::spawn(async move {
        if let Err(e) = fut.await {
            eprintln!("{}", e)
        }
    })
}
